package org.mlf.fsyq.demo.fraud;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.walkthrough.common.sink.AlertSink;
import org.apache.flink.walkthrough.common.entity.Alert;
import org.apache.flink.walkthrough.common.entity.Transaction;

/**
 * FraudDetectionJob 类定义了应用的数据流。
 * @author cjh
 */
public class FraudDetectionJob {


    public static void main(String[] args) throws Exception {

        //这个env对象包含了你Job的属性，构建你的数据源，并最终触发Job的运行。
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //数据源会从现有的数据源系统里读取数据，比如kafka，pulsar等，读取到Flink里。
        //这个案例使用了一个可以产生无限多个信用卡交易记录信息的数据源，让你来处理。
        //每条交易信息包括 id，交易时间戳，交易总金额三个信息。
        //这个数据源的名字[transactions]只是为了测试使用，所以如果哪里出了错误，我们就能第一时间知道了
        //新建输入流,从自定义的Transaction读入
        DataStream<Transaction> transactions = env
                .addSource(new TransactionSource())
                .name("transactions");

        //因为数据源数据非常多，所以它需要被多个欺诈检测程序并行处理。
        //由于欺诈可能发生在每个账户上，所以你必须确保每个账户是被相同的并行欺诈检测器处理的
        //为了确保相同的物理任务处理特定键的所有记录，可以使用DataStream#keyBy对流进行分区
        //process()调用添加一个运算符，该运算符将函数应用于流中的每个分区元素。
        //通常说，紧接在keyBy之后的操作符(在本例中为欺诈检测器)是在一个键控上下文中执行的。
        DataStream<Alert> alerts = transactions
                .keyBy(Transaction::getAccountId)//通过id分组,相同id的流会被分配到同一个slot上
                .process(new FraudDetector())//对每个分组进行欺诈校验
                .name("fraud-detector");

        //sink将数据流写入外部系统;比如Apache Kafka、Cassandra和AWS Kinesis。
        //AlertSink用日志级别的信息记录每个警告记录，而不是将其写入持久存储，因此您可以很容易地看到结果，就是把写在磁盘上的日志信息打印到屏幕上了。
        alerts
                .addSink(new AlertSink())
                .name("send-alerts");
        alerts.print();

        //Flink应用程序是惰性地构建的，并且只在完全形成后才被交付给集群执行。
        env.execute("Fraud Detection");
    }
}
